package com.atguigu;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName: test
 * @Description:
 * @Author: kele
 * @Date: 2021/4/7 18:59
 **/
public class test {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 使用sql从文件读取数据
        tenv.executeSql(
                "create table user_behavior(" +
                        "   user_id bigint, " +
                        "   item_id bigint, " +  //商品id
                        "   category_id int, " + // 品类id
                        "   behavior string, " + // pv cart ...
                        "   ts bigint, " +   // 时间戳
                        "   event_time as to_timestamp(from_unixtime(ts, 'yyyy-MM-dd HH:mm:ss')), " +
                        "   watermark for event_time as  event_time - interval '5' second " +
                        ")with(" +
                        "   'connector'='filesystem', " +
                        "   'path'='in/UserBehavior.csv', " +
                        "   'format'='csv')"
        );

        // 1. 使用滑窗, 计算每个商品每个窗口的点击量
        Table t1 = tenv.sqlQuery("select" +
                "  item_id, " +
                "  hop_end(event_time, interval '10' minute, interval '1' hour) w_end, " +
                "  count(*) item_count " +
                "  from user_behavior " +
                "  where behavior='pv' " +
                "  group by item_id, hop(event_time, interval '10' minute, interval '1' hour)");
        tenv.createTemporaryView("t1", t1);
        // 2. 使用over窗口, 每个点击量排序 rank  dense_rank   row_number
        Table t2 = tenv.sqlQuery("select " +
                "   item_id, " +
                "   w_end, " +
                "   item_count, " +
                "   row_number() over(partition by w_end order by item_count desc) rn" +
                " from t1 ");
        tenv.createTemporaryView("t2", t2);
        // 3. 过滤出top3
        Table t3 = tenv.sqlQuery("select " +
                "   item_id, " +
                "   w_end, " +
                "   item_count, " +
                "   rn " +
                " from t2 " +
                " where rn<=3");
//        tenv.createTemporaryView("t3", t3);
        // 4. 写入到mysql
        // 4.1 创建输出表
        tenv.executeSql("create table hot_item2(" +
                "   item_id bigint, " +
                "   w_end timestamp(3), " +
                "   item_count bigint, " +
                "   rk bigint, " +
                "   PRIMARY KEY (w_end, rk) NOT ENFORCED)" +
                "with(" +
                "   'connector' = 'jdbc', " +
                "   'url' = 'jdbc:mysql://hadoop162:3306/test?useSSL=false', " +
                "   'table-name' = 'hot_item2', " +
                "   'username' = 'root', " +
                "   'password' = 'aaaaaa' " +
                ")");

        t3.executeInsert("hot_item2");

    }
 }

